/*
 * Copyright 2009 Armando Blancas
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package taskgraph.tasks.io;

import java.beans.IntrospectionException;
import java.beans.PropertyEditor;
import java.io.BufferedWriter;
import java.io.EOFException;
import java.io.FileOutputStream;
import java.io.IOException;
import java.io.InterruptedIOException;
import java.io.OutputStreamWriter;
import java.io.PrintWriter;
import java.io.UnsupportedEncodingException;
import java.lang.reflect.InvocationTargetException;
import java.lang.reflect.Method;
import java.nio.charset.Charset;
import java.util.Arrays;

import taskgraph.Config;
import taskgraph.ports.InputPort;
import taskgraph.reflect.Accessor;
import taskgraph.reflect.BeanMetadata;
import taskgraph.tasks.PrimitiveTask;


/**
 * Sink task for writing a delimited text file.
 *  
 * @author Armando Blancas
 * @see taskgraph.ports.InputPort
 * @param <E> The type of elements written by this task.
 */
public class WriteDelimitedTextFile<E> extends PrimitiveTask {

	private static final char   DEFAULT_DELIMITER = ',';
	private static final char   DEFAULT_QUALIFIER = 0;
	private static final String DEFAULT_ENCODING  = Charset.defaultCharset().name();
	private static final int    DEFAULT_CAPACITY  = Config.get().bufferCapacity();
	
	private InputPort<E> input;
	private String filename;
	private String charsetName = DEFAULT_ENCODING;
	private char delimiter = DEFAULT_DELIMITER;
	private char qualifier = DEFAULT_QUALIFIER;
	private boolean appending;
	private boolean writingHeaders;
	private int bufferSize = DEFAULT_CAPACITY;

	/**
	 * This constructor allows creating instances as beans.
	 */
    public WriteDelimitedTextFile() {
    	// do nothing
    }

	/**
	 * WriteDelimitedTextFile constructor.
	 * 
     * Creates a fully configured instance, with default delimiter
     * and qualifier characters.
     * 
     * @param input     The element output port.
     * @param filename  The full path of the file to read as String.
    */
    public WriteDelimitedTextFile(final InputPort<E> input, 
    		                      final String filename) {
    	setInput(input);
    	setFilename(filename);
    }

	/**
     * Creates a fully configured instance.
     * 
     * @param input     The element output port.
     * @param filename  The full path of the file to read as String.
     * @param charsetName  The name of the character encoding as String.
     * @param delimiter The character that separates fields in the file.
     * @param qualifier The character that surrounds fields in the file.
	 * @throws UnsupportedEncodingException 
     */
    public WriteDelimitedTextFile(final InputPort<E> input,
    	                          final String filename,
     	                          final String charsetName,
    	                          final char delimiter,
    	                          final char qualifier)
    throws UnsupportedEncodingException {
    	setInput(input);
    	setFilename(filename);
    	setCharsetName(charsetName);
    	setDelimiter(delimiter);
    	setQualifier(qualifier);
    }

	/**
	 * Calls the {@code getter} method on the {@code obj} argument.
	 * Then uses a bean property editor to convert the returned value
	 * to its proper text representation, which is then put inside
	 * qualifier characters.
	 * 
	 * @param obj      The object that is being queried for values.
	 * @param accessor An accessor that wraps a getter method.
	 * @return         A text value inside a qualifier character, 
	 *               usually the double quote.
	 * @throws IllegalArgumentException
	 * @throws IllegalAccessException
	 * @throws InvocationTargetException
	 */
	private String formatField(Object obj, Accessor accessor) 
	throws IllegalArgumentException, 
	       IllegalAccessException, 
	       InvocationTargetException {
		String field = "";
		Method getter = accessor.getMethod();
		Class<?> paramType = getter.getReturnType();
		PropertyEditor editor = BeanMetadata.editorFor(paramType);
		if (editor != null) {
			editor.setValue(getter.invoke(obj));
			if (qualifier == 0) {
				field = editor.getAsText();
			}
			else {
				field = String.format("%c%s%c", qualifier, editor.getAsText(), qualifier);
			}
		}
		return field;
	}

	/**
	 * Gets the buffer size for reading chunks of characters.
	 * 
	 * @return the bufferSize
	 */
	public int getBufferSize() {
		return bufferSize;
	}

	/**
	 * Gets the name of the character encoding.
	 * 
	 * @return The charsetName value as {@code String}.
	 */
	public String getCharsetName() {
		return charsetName;
	}

    /**
	 * Gets the character that separates field values.
	 * 
	 * @return The delimiter {@code char} value.
	 */
	public char getDelimiter() {
		return delimiter;
	}

	/**
     * Gets the name of the file to read.
     * 
	 * @return The full path of the file as String
	 */
	public String getFilename() {
		return filename;
	}

	/**
     * Gets the input port.
     * 
	 * @return The element input port.
	 */
	public InputPort<E> getInput() {
		return input;
	}

	/**
	 * Gets the character that may surround field values.
	 * 
	 * @return The qualifier {@code char} value.
	 */
	public char getQualifier() {
		return qualifier;
	}
	
	/**
	 * Tests whether this task is appending data to an existing file.
	 * <p>If {@code false} the file will be overwritten.
	 * 
	 * @return the appending
	 */
	public boolean isAppending() {
		return appending;
	}

	/**
	 * Tests whether this task is writing field headers.
	 * 
	 * @return A {@code boolean} value.
	 */
	public boolean isWritingHeaders() {
		return writingHeaders;
	}

	/**
     * Performs the work of this task.
     * 
	 * Writes a character representation of element objects coming from the
	 * input port into an output file. Reach record occupies a line of text;
	 * fields are bounded by a delimiter character; fields that contain the
	 * delimiter are surrounded by a qualifier character.
	 *<pre>
	 *open text file
	 *while not EOF
	 *    get an element E from the input port
	 *    print each field and add the delimiter
	 *    print a newline separator.
	 *close text file
	 *close input port
	 *</pre>
	 */
	@Override
    public void run() {
		PrintWriter output = null;
		try {
    		output = new PrintWriter(
    				 	 new BufferedWriter(
    				 		 new OutputStreamWriter(
    				 			 new FileOutputStream(filename, appending),
    				 			 charsetName),
    				 		 bufferSize),
    				 	false);
    		
    		E element = input.read();
    		Class<?> clazz = element.getClass();
			Accessor[] getters = BeanMetadata.extractGetters(clazz);
			if (getters.length > 0) {
				Arrays.sort(getters);
				if (isWritingHeaders()) {
					output.print(getters[0].getPropertyName());
    				for (int i = 1; i < getters.length; ++i) {
    					output.append(delimiter)
    					      .append(getters[i].getPropertyName());
    				}
    				output.println();					
				}
				while (!Thread.currentThread().isInterrupted()) {
    				output.print(formatField(element, getters[0]));
    				for (int i = 1; i < getters.length; ++i) {
    					output.append(delimiter)
    					      .print(formatField(element, getters[i]));
    				}
    				output.println();
					element = input.read();
				}
			} else {
				log.warn("WriteDelimitedTextFile#run(): No getter methods found.");
			}
        } catch (InterruptedIOException e) {
        	// interruption request
        } catch (EOFException e) {
        	// we're done reading
        } catch (IOException e) {
            log.error("WriteDelimitedTextFile#run()", e);
		} catch (IntrospectionException e) {
            log.error("WriteDelimitedTextFile#run()", e);
		} catch (IllegalArgumentException e) {
            log.error("WriteDelimitedTextFile#run()", e);
		} catch (IllegalAccessException e) {
            log.error("WriteDelimitedTextFile#run()", e);
		} catch (InvocationTargetException e) {
            log.error("WriteDelimitedTextFile#run()", e);
		} finally {
        	input.close();
        	if (output != null)
        		output.close();
        }
    }

	/**
	 * Sets whether this task is appending data to an existing file.
	 * <p>{@code false} means the file will be overwritten.
	 * 
	 * @param appending The {@code boolean} value to set.
	 */
	public void setAppending(boolean appending) {
		this.appending = appending;
	}

	/**
	 * Sets the buffer size for reading chunks of characters.
	 * <p>The argument should not be less than the default. If it is, the
	 * default size is used instead.
	 * 
	 * @param bufferSize the bufferSize to set
	 */
	public void setBufferSize(final int bufferSize) {
		this.bufferSize = (bufferSize < DEFAULT_CAPACITY) ? DEFAULT_CAPACITY : bufferSize;
	}

	/**
	 * Sets the name of the character encoding.
	 * 
	 * @param charsetName The {@code String} value to set as charset name.
	 * @throws UnsupportedEncodingException 
	 */
	public void setCharsetName(final String charsetName)
	throws UnsupportedEncodingException {
    	if (charsetName == null)
    		throw new IllegalArgumentException("charsetName == null");
		if (!Charset.isSupported(charsetName)) {
			throw new UnsupportedEncodingException (charsetName);
		}
		this.charsetName = charsetName;
	}

	/**
	 * Sets the character that separates field values.
	 * 
	 * @param delimiter The delimiter {@code char} value.
	 */
	public void setDelimiter(char delimiter) {
		this.delimiter = delimiter;
	}

	/**
	 * Sets the name of the file to read.
	 * 
	 * @param filename The full path of the file as String.
	 */
	public void setFilename(final String filename) {
    	if (filename == null)
    		throw new IllegalArgumentException("filename == null");
		this.filename = filename;
	}

	/**
	 * Sets the input port.
	 * 
	 * @param input The element input port.
	 */
    public void setInput(final InputPort<E> input) {
    	if (input == null)
    		throw new IllegalArgumentException("input == null");
 		this.input = input;
	}

	/**
	 * Sets the character that may surround field values.
	 * 
	 * @param qualifier The qualifier {@code char} value.
	 */
	public void setQualifier(char qualifier) {
		this.qualifier = qualifier;
	}

	/**
	 * Sets whether this task is writing field headers.
	 * 
	 * @param writingHeaders The {@code boolean} value to set.
	 */
	public void setWritingHeaders(boolean writingHeaders) {
		this.writingHeaders = writingHeaders;
	}

}
